package plagiarismChecker;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;


public class DataReader {

	/**
	 * Holds writer instance.
	 * Every time writer is called new file is created not appended to last one.
	 * 
	 * ! writer.close() has to be called when data is written
	 */
	private static SequenceFile.Writer writer = null;
	
	/**
	 * This is used to close writer at the right time
	 */
	private static int linesWritten = 0;

	/**
	 * Writes mapreduce data to it's input
	 * 
	 * @param key
	 * @param value
	 * @param totalLines total number of lines. This is used to close writer at right time
	 * @param pathStr where to write
	 * @param conf
	 * @throws IOException
	 */
	public static void writeData (String key, String value, int totalLines, String pathStr, Configuration conf) throws IOException {
		if (writer == null) {
			FileSystem fs = FileSystem.get(conf);
			Path path = new Path(pathStr);
			writer = SequenceFile.createWriter(fs, conf, path, Text.class, Text.class, SequenceFile.CompressionType.NONE);
		}
		Text index = new Text(key);
		Text num = new Text(value);
		writer.append(index, num);
		closeWhenLastWritten(totalLines);
	}
	
	/**
	 * Reads mapreduce data
	 * 
	 * @param pathStr where to read
	 * @param conf Configuration
	 * @return data line by line
	 * @throws IOException
	 */
	public static ArrayList<HashMap<String, String>> readData (String pathStr, Configuration conf) throws IOException {
		ArrayList<HashMap<String, String>> data = new ArrayList<HashMap<String, String>>();
		
		FileSystem fs = FileSystem.get(conf);
		Path path = new Path(pathStr);
		
		if (!fs.isFile(path)) {
			FileStatus[] listFiles = fs.listStatus(path);
			for (int i = 0; i < listFiles.length; i++) {
				try {
					SequenceFile.Reader reader = new SequenceFile.Reader(fs, listFiles[i].getPath(), conf);
					Text index = new Text();
					Text el = new Text();
					while (reader.next(index, el)) {
						HashMap<String, String> map = new HashMap<String, String>();
						map.put("key", index.toString());
						map.put("value", el.toString());
						data.add(map);
					}
					reader.close();
				}
				catch(Exception e){

				}
			}
		}
		return data;
	}
	
	/**
	 * Closes writer
	 */
	private static void closeWhenLastWritten(int totalLines) {
		linesWritten ++;
		if (linesWritten == totalLines) {
			try {
				writer.close();
				writer = null;
				linesWritten = 0;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	
}